package com.shengzai.flink.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

object Demo5MySource {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val mySourceDS: DataStream[String] = env.addSource(new MySource)

    mySourceDS.print()

    env.execute()

  }

}
class MySource extends SourceFunction[String]{
  /**
   *
   * @param ctx:发送到下游的文本
   */
  override def run(ctx: SourceFunction.SourceContext[String]): Unit = {

    while (true){
      ctx.collect("java")
      Thread.sleep(1000)
    }

  }

    //关闭时的方法，一般用于关闭资源
  override def cancel():Unit={}

}
